﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Ensemble;
using System.Runtime.Serialization.Formatters.Binary;
using System.IO;
using System.Threading;
using System.ServiceModel;
using FlightSearchServer;

namespace TicketSeller
{
    class ClusterCommunicator
    {
        private Connection connection   = new Connection();

        private JoinOps jops            = new JoinOps();

        private Member member;

        private string nodeName;

        private string alliance;

        private const int REPLICA_COPIES = 1;

        private RegistrationServiceClient proxy;

        private bool isRegistered = false;

        public ClusterCommunicator( string nodeName, string allianceName, string searchServerAddr, int portNo)
        {
            this.nodeName = nodeName;
            this.alliance = allianceName;
            this.portNo = portNo;
            // Setup connection
            connection.Connect();
            jops.group_name = allianceName;
            member = new Member(connection);
            member.Join(jops);
            EndpointAddress address = new EndpointAddress(new Uri( searchServerAddr));
            proxy = new RegistrationServiceClient(new BasicHttpBinding(), address);
            ServersCollection list = new ServersCollection();
            try {
                proxy.register( String.Format("http://localhost:{0}/FlightSearch/{1}", portNo, nodeName),
                    nodeName, alliance, list);
                isRegistered = true;
            }
            catch (EndpointNotFoundException e)
            {
                throw new RegistrationServerDown( );
            }
        }

        public void serveCommunication()
        {
            int viewCount = 0;
            IList<ServerNode> availableNodes = new List<ServerNode>();
            ClusterManager.Instance.NodeName = nodeName;
            while (true)
            {
                // Read all waiting messages from Ensmeble
                while (connection.Poll())
                {
                    Message msg = connection.Recv();
                    switch (msg.mtype)
                    {
                        case UpType.VIEW:
                            lock (connection)
                            {
                                try
                                {
                                    if (member.current_status == Member.Status.Normal)
                                    {
                                        viewCount = member.current_view.nmembers;
                                        availableNodes.Clear();
                                        // Send an introduction message to all nodes.
                                        int rank = member.current_view.rank;
                                        ServerNode node = new ServerNode( nodeName, rank);
                                        foreach (string serverName in ClusterManager.Instance.ReplicatedAt)
                                            node.replicateAt(serverName);
                                        availableNodes.Add(node);
                                        ClusterManager.Instance.registerNode(node);
                                        broadCastMessage( new ClusterMessage( node, MessageType.INTRODUCTION));
                                        Console.WriteLine( "View count: {0}", viewCount);
                                    }
                                    // Only if node received 0 as rank means it became a leader
                                    // of nodes cluster.
                                    ClusterManager.Instance.IsLeader = member.current_view.rank == 0;
                                    if (viewCount == 1)
                                    {
                                        ClusterManager.Instance.calculateSuspected( availableNodes);
                                    }
                                }
                                catch (Exception e)
                                {
                                    Console.WriteLine(e.Message);
                                }
                            }
                            break;
                        case UpType.CAST:
                            {
                                if (msg.data == null)
                                    break;
                                MemoryStream ms = new MemoryStream(msg.data);
                                BinaryFormatter formatter = new BinaryFormatter();
                                ClusterMessage m = (ClusterMessage)formatter.Deserialize(ms);

                                switch (m.Type)
                                {
                                    case MessageType.INTRODUCTION:
                                        {
                                            ServerNode n = m.Node;
                                            // Phase I. All nodes should be introduced to each one.
                                            viewCount--;
                                            availableNodes.Add(m.Node);
                                            ClusterManager.Instance.registerNode(n);
                                            if (viewCount != 1)
                                                break;
                                            ClusterManager.Instance.calculateSuspected( availableNodes);
                                            // Phase II. After introduction has been finished, need
                                            // to decide where and how to replicate data.
                                            replicateCurrentNode( availableNodes);
                                            // Phase III.
                                            // Here need to take care of nodes for which we have replications
                                            // but nodes itself failed down.
                                            replicateFailedNodes( availableNodes);
                                            ServersCollection collection = new ServersCollection();
                                            foreach (String server in ClusterManager.Instance.ReplicatedAt)
                                            {
                                                SearchServer searchServer = new SearchServer();
                                                searchServer.Name = server;
                                                searchServer.Alliance = alliance;
                                                collection.Add(searchServer);
                                            }
                                            if (!isRegistered)
                                            {
                                                proxy.register(String.Format("http://localhost:{0}/FlightSearch/{1}", portNo, nodeName),
                                                    nodeName, alliance, collection);
                                            }
                                            else
                                            {
                                                proxy.update(nodeName, alliance, collection);
                                            }
                                            break;
                                        }
                                    default:
                                        break;
                                }

                                break;
                            }
                        case UpType.SEND:
                        {
                                if (msg.data == null)
                                    break;
                                MemoryStream ms = new MemoryStream(msg.data);
                                BinaryFormatter formatter = new BinaryFormatter();
                                ClusterMessage m = (ClusterMessage)formatter.Deserialize(ms);
                                switch (m.Type)
                                {
                                    case MessageType.REPLICATION:
                                        {
                                            ServerNode srvNode = m.Node;
                                            Console.WriteLine("Name: {0}, tickets count: ({1}) ", srvNode.Name, srvNode.Tickets.Count());
                                            ClusterManager.Instance.HasReplicationOf.Add(srvNode.Name);
                                            ClusterManager.Instance.updateClusterNode( srvNode);
                                            foreach (Ticket ticket in srvNode.Tickets)
                                            {
                                                TicketsDB.Instance.insertReplication(srvNode.Name, ticket);
                                            }
                                            break;
                                        }
                                    default:
                                        break;
                                }
                                break;
                         }
                        case UpType.BLOCK:
                            member.BlockOk();
                            break;
                        case UpType.EXIT:
                            break;
                    }
                    Console.Out.Flush();
                }
                //Console.WriteLine("sleeping for a little while");
                Thread.Sleep(100);
            }
        }

        private void broadCastMessage( ClusterMessage message)
        {
            MemoryStream stream = new MemoryStream();
            BinaryFormatter formatter = new BinaryFormatter();

            formatter.Serialize(stream, message);

            member.Cast(stream.ToArray());
        }

        private void sendMessage(ClusterMessage message, int[] ids)
        {
            MemoryStream stream = new MemoryStream();
            BinaryFormatter formatter = new BinaryFormatter();

            formatter.Serialize(stream, message);
            member.Send(ids, stream.ToArray());
        }

        private void replicateCurrentNode(IList<ServerNode> availableNodes)
        {
            ClusterManager.Instance.calculateSuspected(availableNodes); // Recalculate each time list of suspected/failed nodes

            // Compute (replicatedAt - suspectedNodes) set
            var replicationAlive = (from replica in ClusterManager.Instance.ReplicatedAt // Replication of current node which alive
                                    where !(from failed in ClusterManager.Instance.SuspectedNodes select failed).Contains(replica)
                                    select replica).ToList();
            if (replicationAlive.Count >= REPLICA_COPIES ||
                 replicationAlive.Count >= availableNodes.Count) // Minimum of replicated storage is alive no need to replicate
                return;
            int replicaCount = replicationAlive.Count;
            while (replicaCount < REPLICA_COPIES)
            {
                // Have replication at each node already.
                if (replicaCount == availableNodes.Count)
                    break;
                Random rnd = new Random();
                int index = Convert.ToInt32(Math.Floor(rnd.NextDouble() * (availableNodes.Count )));
                ServerNode node = availableNodes.ElementAt(index);
                if (ClusterManager.Instance.ReplicatedAt.Contains(node.Name) || nodeName == node.Name)
                    continue;
                ClusterManager.Instance.ReplicatedAt.Add(node.Name); // Replicate current node to selected one since old replication down
                replicaCount++;
            }
            var selected = from t in availableNodes
                           where (from t2 in ClusterManager.Instance.ReplicatedAt select t2).Contains(t.Name)
                           select t.Rank; 
            Console.WriteLine("Replication");

            foreach (string str in (from nodeN in availableNodes
                                    where (from t in selected select t).Contains(nodeN.Rank)
                                    select nodeN.Name))
            {
                // Need to be removed in release
                Console.WriteLine("Send replica to: {0}", str);
            }
            ServerNode selfNode = new ServerNode();
            selfNode.Rank = member.current_view.rank;
            selfNode.Name = nodeName;
            selfNode.Tickets = TicketsDB.Instance.getTicketsCopy();
            foreach (string name in ClusterManager.Instance.ReplicatedAt)
                selfNode.replicateAt(name);
            // Update everyone who keeps my replication

            sendMessage(new ClusterMessage(selfNode, MessageType.REPLICATION), selected.ToArray());
            // Need to add server registration/update
        }

        private void replicateFailedNodes(IList<ServerNode> availableNodes)
        {
            foreach (String node in ClusterManager.Instance.HasReplicationOf)
            {
                var nodeAlive = (from n in availableNodes
                                 where n.Name == node
                                 select n).ToList().Count > 0;
                // Check whenever node itself alive
                // and in case it's there is nothing we should do
                // since node will take care of replication.
                if (nodeAlive)
                    continue;
                var replicatedNode = (from n in ClusterManager.Instance.ClusterNodes
                                     where n.Name.Equals( node)
                                     select n).ToList( ).ElementAt( 0);
                /* ============================================================== */
                var isLowestRank = (from n in availableNodes
                                    where replicatedNode.ReplicatedAt.Contains( n.Name)
                                    orderby n.Rank
                                    select n.Rank).ToList( ).ElementAt(0) == member.current_view.rank;
                if ( !isLowestRank)
                    continue;
                // Now need to check how much replications do we have alive?

                var replicationNum = ( from n in availableNodes
                                       where (from n1 in replicatedNode.ReplicatedAt select n1).Contains( n.Name)
                                       select n).ToList( ).Count;
                if ( replicationNum >= (REPLICA_COPIES + 1) ||
                    replicationNum >= availableNodes.Count)
                    continue;
                // Well the server is down we have
                // smallest id and there is not enought replications, 
                // hence need to care keep
                // correct number of replications in the alliance/cluster
                while ( replicationNum < (REPLICA_COPIES + 1)) {
                    if (replicationNum == availableNodes.Count)
                        break; // Do not have enought available node to keep all replications.
                    Random rnd = new Random( );
                    int index = Convert.ToInt32(Math.Floor(rnd.NextDouble() * (availableNodes.Count)));
                    ServerNode tmp = availableNodes.ElementAt(index);
                    if (replicatedNode.ReplicatedAt.Contains(tmp.Name) || nodeName == tmp.Name)
                        continue;
                    replicatedNode.replicateAt( tmp.Name);
                    ++replicationNum;
                }
                var selected = ( from n in availableNodes
                                 where ( replicatedNode.ReplicatedAt.Contains( n.Name))
                                 select n.Rank);
                replicatedNode.Tickets = TicketsDB.Instance.getTicketsByServerName( replicatedNode.Name);
                Console.WriteLine( "Going to replicate {0}, current rank is {1}", replicatedNode.Name, member.current_view.rank);
                sendMessage(new ClusterMessage(replicatedNode, MessageType.REPLICATION), selected.ToArray());
            }
        }

        static string generateString(int size)
        {
            StringBuilder builder = new StringBuilder();
            Random rnd = new Random();

            for (int i = 0; i < size; ++i)
            {
                int toUpper = (rnd.NextDouble() > 0.5) ? 97 : 65;
                builder.Append(Convert.ToChar(Convert.ToInt32(Math.Floor(26 * rnd.NextDouble() + toUpper))));
            }
            return builder.ToString();
        }

        public int portNo { get; set; }
    }
}
